package cc.eddic.practice.concurrency.impl;

import cc.eddic.practice.concurrency.CDNLogItem;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import lombok.var;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.zip.GZIPInputStream;


@Slf4j
public class CDNLogParserWithFutureCxm {
    private static final int NUM_THREADS = Runtime.getRuntime().availableProcessors();

    public Map<String, List<CDNLogItem>> parse(String folderName) {
        Map<String, List<CDNLogItem>> res = new ConcurrentHashMap<>();
        ExecutorService executorService = Executors.newFixedThreadPool(NUM_THREADS);
        try {
            val folderPath = Paths.get(folderName);
            try (var walk = Files.walk(folderPath)) {
                List<CompletableFuture<Model>> futures = walk.parallel()
                        .filter(Files::isRegularFile)
                        .filter(path -> path.getFileName().toString().endsWith(".gz"))
                        .map(path -> CompletableFuture.supplyAsync(() -> parseFile(path), executorService))
                        .collect(Collectors.toList());

                CompletableFuture<Void> allFutures = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
                allFutures.join();

                futures.stream()
                        .map(CompletableFuture::join)
                        .filter(Objects::nonNull)
                        .forEach(model -> res.put(model.getPath().toString(), model.getItems()));
            }
        } catch (IOException e) {
            log.error("File does not exist: {}", e.getMessage());
        }

        return res;
    }

    private Model parseFile(Path path) {
        try (var gzipStream = new GZIPInputStream(Files.newInputStream(path));
             var reader = new BufferedReader(new InputStreamReader(gzipStream))) {
             Stream<String> fileLines=reader.lines();
             val items = fileLines.map(new CDNLogParserCxm()::parse)
                    .filter(Optional::isPresent)
                    .map(Optional::get)
                    .collect(Collectors.toList());

             if (log.isDebugEnabled()) {
                log.debug("PARSE: {} => {}", path, items.size());
             }

            return new Model(path, items);
        } catch (IOException ex) {
            log.error(ex.getMessage(), ex);
        }

        return new Model(path);
    }

    @Getter
    private static class Model {

        private final Path path;
        private final List<CDNLogItem> items;

        public Model(Path path) {
            this(path, new ArrayList<>());
        }

        public Model(Path path, List<CDNLogItem> logItems) {
            this.path = path;
            this.items = logItems;
        }

        public Path getPath() {
            return path;
        }

        public List<CDNLogItem> getItems() {
            return items;
        }
    }
}

